#include <QUdpSocket>
#include <QTcpSocket>
#include <QTcpServer>
#include "QQuickROChannel.h"

struct QQuickROChannel::ServerData
{
    QQuickROChannelServer* m_server = nullptr;
};

struct QQuickROChannel::ClientData
{
    QUdpSocket* m_udpSocket = nullptr;
    QTimer* connectTimer = nullptr;
};

struct DataPacket
{
    QString topic;
    QVariant value;
};

static const QString flag = QStringLiteral("IIXI");

static void sendUdp(QUdpSocket* socket,QHostAddress address,int port,int conId,const DataPacket& packet){


    QByteArray data;

    {
        QDataStream ss(&data,QIODevice::WriteOnly);
        ss << flag << conId << packet.topic << packet.value;
    }

    data = qCompress(data);

    socket->writeDatagram(data,address,port);
}

static bool readUdp(QUdpSocket* socket,QHostAddress &address,quint16& port,int& conId,DataPacket& packet){
    QByteArray data;
    data.resize(socket->pendingDatagramSize());
    socket->readDatagram(data.data(),data.size(),&address,&port);
    data = qUncompress(data);
    {
        QDataStream ss(&data,QIODevice::ReadOnly);
        QString f;
        ss >> f >> conId >> packet.topic >>packet.value;
        if(f != flag)
        {
            return false;
        }
    }
    return true;
}

static void sendTcp(QTcpSocket* socket,const DataPacket& packet){
    QByteArray data;
    QDataStream s(&data,QIODevice::WriteOnly);
    s << packet.topic << packet.value;
    data = qCompress(data);
    int size = data.size();
    socket->write((char*)&size,sizeof(size));
    socket->write(data);
    QMetaObject::invokeMethod(socket,[=](){
            socket->flush();//延后发送包，降低网络压力
    },Qt::QueuedConnection);
}

static bool readTcp(QTcpSocket* socket,DataPacket& packet){
    int size = 0;
    if(socket->bytesAvailable() < sizeof(size))
    {
        return false;
    }

    socket->peek((char*)&size,sizeof(size));
    if(socket->bytesAvailable() < sizeof(size) + size){
        return false;
    }

    socket->skip(sizeof(size));
    auto data = socket->read(size);
    data = qUncompress(data);
    QDataStream s(&data,QIODevice::ReadOnly);
    s >> packet.topic >> packet.value;
    return true;
}

QQuickROChannel::QQuickROChannel(QObject *parent):QObject(parent)
{




}

QQuickROChannel::~QQuickROChannel()
{
    if(m_sData)
    {
        delete m_sData;
    }

    if(m_cData)
    {
        delete m_cData;
    }
}

void QQuickROChannel::connectTo(QHostAddress address, int port)
{
    setState(ConnectingToHost);
    m_cData = new ClientData();
    m_tcpSocket = new QTcpSocket(this);

    connect(m_tcpSocket,&QTcpSocket::readyRead,this,&QQuickROChannel::onTcpReadyRead);
    connect(m_tcpSocket,&QTcpSocket::disconnected,this,&QQuickROChannel::onTcpDisconnected);

    m_tcpSocket->connectToHost(address,port);
}

void QQuickROChannel::disconnect()
{
    if(m_state == Ready)
    {
        m_tcpSocket->disconnectFromHost();
    }
    else{
        qDebug()<<"disconnect failed channel not ready";
    }
}

void QQuickROChannel::write(QString topic, QVariant value, MessagePolicy policy)
{
    if(m_state != Ready)
    {
        qDebug()<<__func__<<"ChannelNotReady";
    }
    if(policy == Safe)
    {
        sendTcp(m_tcpSocket,{topic,value});
    }
    else if(policy == Quick){
        QUdpSocket* udpSocket = nullptr;

        if(m_sData)
        {
            udpSocket = m_sData->m_server->m_udpServer;
        }
        else if(m_cData)
        {
            udpSocket = m_cData->m_udpSocket;
        }

        sendUdp(udpSocket,m_peerUdpAddress,m_peerUdpPort,conId,{topic,value});
    }
}

QQuickROChannel::State QQuickROChannel::state()
{
    return m_state;
}

int QQuickROChannel::id()
{
    return conId;
}

void QQuickROChannel::setState(State state)
{
    if(state == m_state)
    {
        return;
    }

    m_state = state;
    emit stateChanged();
}

void QQuickROChannel::connectToClient(QQuickROChannelServer* server,QTcpSocket *socket, int conId)
{
    setState(ConnectingToClient);
    m_sData = new ServerData();
    m_tcpSocket = socket;
    m_sData->m_server = server;
    this->conId = conId;
    socket->setParent(this);
    connect(socket,&QTcpSocket::readyRead,this,&QQuickROChannel::onTcpReadyRead);
    connect(socket,&QTcpSocket::disconnected,this,&QQuickROChannel::onTcpDisconnected);

    //客户端第一次链接时服务端会主动发送注册信息.内部包含客户端的连接id，以及udp端口号
    sendTcp(socket,{"channelInfo",getChannelInfo()});
}

void QQuickROChannel::onMessage(QString topic, QVariant value,const SenderInfo& source)
{
    if(topic == "clientReady")//自身是服务端而对方是客户端，此时客户端完成了连接
    {
        if(m_state == ConnectingToClient)
        {
            m_peerUdpAddress = source.udpaddress;
            m_peerUdpPort = source.udpport;//记录对方的端口
            sendTcp(m_tcpSocket,{"serverReady",{}});//向客户端发送服务端链接成功信号
            setState(Ready);
        }
    }
    else if(topic == "channelInfo")//自身是客户端而对方是服务端，此时服务端发来了注册信息
    {
        if(m_state == ConnectingToHost)
        {
            auto param = value.toMap();
            m_peerUdpAddress = m_tcpSocket->peerAddress();
            m_peerUdpPort = param["udpPort"].toInt();
            conId = param["channelId"].toInt();
            m_cData->m_udpSocket = new QUdpSocket(this);
            m_cData->m_udpSocket->bind(QHostAddress::Any);
            connect(m_cData->m_udpSocket,&QUdpSocket::readyRead,this,&QQuickROChannel::onUdpReadyRead);


            m_cData->connectTimer = new QTimer(this);
            auto send = [=](){
                sendUdp(m_cData->m_udpSocket,m_tcpSocket->peerAddress(),m_peerUdpPort,conId,{"clientReady",{}});
            };
            connect(m_cData->connectTimer,&QTimer::timeout,this,send);
            m_cData->connectTimer->start(100);
            send();
        }
    }
    else if(topic == "serverReady")
    {
        if(m_state == ConnectingToHost)//自身是客户端而对方是服务端，此时服务端表示连接已完毕
        {
            delete m_cData->connectTimer;
            m_cData->connectTimer = nullptr;
            setState(Ready);
        }
    }
    else{
        emit newMessage(topic,value);
    }
}

void QQuickROChannel::onTcpReadyRead()
{
    DataPacket packet;
    while(readTcp(m_tcpSocket,packet))
    {
        onMessage(packet.topic,packet.value,{SenderInfo::Tcp});
    }
}

void QQuickROChannel::onTcpDisconnected()
{
    setState(Disconnected);
}

void QQuickROChannel::onUdpReadyRead()
{
    DataPacket packet;
    QHostAddress address;
    quint16 port;
    int tempConId;
    while (m_cData->m_udpSocket->hasPendingDatagrams()) {
        if(readUdp(m_cData->m_udpSocket,address,port,tempConId,packet))
        {
            if(tempConId == conId)
            {
                onMessage(packet.topic,packet.value,{SenderInfo::Udp,address,port});
            }
        }
    }
}

QVariant QQuickROChannel::getChannelInfo()
{
    QVariantMap ret = {{"channelId",conId},{"udpPort",m_sData->m_server->m_udpServer->localPort()}};

    return ret;
}

QQuickROChannelServer::QQuickROChannelServer(QObject *parent):QObject(parent)
{
    m_tcpServer = new QTcpServer(this);
    connect(m_tcpServer,&QTcpServer::newConnection,this,[=](){
        while (m_tcpServer->hasPendingConnections()) {
            auto socket = m_tcpServer->nextPendingConnection();
            onNewTcpConnection(socket);
        }

    });

    m_udpServer = new QUdpSocket(this);
    connect(m_udpServer,&QUdpSocket::readyRead,this,[=](){
        DataPacket packet;
        QHostAddress address;
        quint16 port;
        int tempConId;
        while (m_udpServer->hasPendingDatagrams()) {
            if(readUdp(m_udpServer,address,port,tempConId,packet))
            {
                auto c = m_channels.value(tempConId);

                if(!c)
                {
                    c = m_tempChannels.value(tempConId);
                }

                if(c)
                {
                    c->onMessage(packet.topic,packet.value,{QQuickROChannel::SenderInfo::Udp,address,port});
                }
            }
        }

    });
}

bool QQuickROChannelServer::listen(QHostAddress address, int port)
{
    if(!m_tcpServer->listen(address,port))
    {
        return false;
    }

    if(!m_udpServer->bind(address)){
        return false;
    }


    return true;
}

QList<QQuickROChannel *> QQuickROChannelServer::channels() const
{
    return m_channels.values();
}

void QQuickROChannelServer::onNewTcpConnection(QTcpSocket *socket)
{
    QQuickROChannel* channel = new QQuickROChannel(this);
    auto conId = allocConId();
    channel->connectToClient(this,socket,conId);

    connect(channel,&QQuickROChannel::stateChanged,this,[=](){
        if(channel->state() == QQuickROChannel::Ready)
        {
            m_channels[conId] = channel;
            m_tempChannels.remove(conId);
            emit newConnection(channel);
        }
    });

    connect(channel,&QObject::destroyed,this,[=](){
        m_tempChannels.remove(conId);
        m_channels.remove(conId);
    });

    m_tempChannels[conId] = channel;
}

int QQuickROChannelServer::allocConId()
{
    maxConId++;
    return maxConId;
}
